package cn.itcast.cep;

import cn.itcast.bean.Message;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

//用户如果在10s内，输入了TMD 5次，就认为用户为恶意攻击，识别出该用户
//使用 Flink CEP检测刷屏用户
public class MaliceUser {
    public static void main(String[] args) throws Exception {
        /**
         *1.获取流处理执行环境
         *  2.设置事件时间
         *  3.构建数据源
         *  4.定义模式规则
         *  5.将规则应用到数据流
         *  6.获取符合规则的数据
         *  7.打印查询到的数据
         *  8.执行任务
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Message> sourceDate = env.fromCollection(Arrays.asList(
                new Message("1", "TMD", 1558430842000L),//2019-05-21 17:27:22
                new Message("1", "TMD", 1558430843000L),//2019-05-21 17:27:23
                new Message("1", "TMD", 1558430845000L),//2019-05-21 17:27:25
                new Message("1", "TMD", 1558430850000L),//2019-05-21 17:27:30
                new Message("1", "TMD", 1558430851000L),//2019-05-21 17:27:30
                new Message("2", "TMD", 1558430851000L),//2019-05-21 17:27:31
                new Message("1", "TMD", 1558430852000L)//2019-05-21 17:27:32
        )).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Message>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Message message) {
                return message.getEventTime();
            }
        });

        //4.定义模式规则
        Pattern<Message, Message> pattern = Pattern.<Message>begin("begin")//设置模式名称
                .where(new IterativeCondition<Message>() {
                    @Override
                    public boolean filter(Message message, Context<Message> context) throws Exception {
                        return message.getMsg().equals("TMD");
                    }
                }).times(5)  //匹配5次
                .within(Time.seconds(10));//窗口时间
        //5.将规则应用到数据流
        PatternStream<Message> sep = CEP.pattern(sourceDate.keyBy(Message::getUserId), pattern);
        //6.获取符合规则的数据
        sep.select(new PatternSelectFunction<Message, Object>() {
            @Override
            public Object select(Map<String, List<Message>> map) throws Exception {
                //map的key为模式名称 ，value是匹配到的数据
                List<Message> begin = map.get("begin");

                return begin;
            }
        }).print();
        //7.打印查询到的数据
        //8.执行任务
        env.execute();
    }
}
